package com.nota.flink.cdc.utils;

import cn.hutool.extra.spring.SpringUtil;
import com.nota.flink.cdc.pojo.DataChangeInfo;
import com.nota.flink.cdc.pojo.constants.RedisStreamConstants;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.alter.Alter;
import net.sf.jsqlparser.statement.alter.AlterExpression;
import net.sf.jsqlparser.statement.alter.AlterOperation;
import net.sf.jsqlparser.statement.create.table.ColDataType;
import net.sf.jsqlparser.statement.create.table.ColumnDefinition;
import net.sf.jsqlparser.statement.create.table.CreateTable;
import redis.clients.jedis.UnifiedJedis;
import redis.clients.jedis.search.IndexDefinition;
import redis.clients.jedis.search.IndexOptions;
import redis.clients.jedis.search.Schema;

import java.io.StringReader;
import java.util.List;

/**
 * @author 胡超雄
 * @date 2024/01/03 11:46
 * @desc sql解析
 */
@NoArgsConstructor
@Slf4j
public class SqlParserUtil {


    public static Boolean parseSql(DataChangeInfo dataChangeInfo) {
        boolean isSuc = false;
        String sql = dataChangeInfo.getDdl();
        try {
            Statement statement = CCJSqlParserUtil.parse(new StringReader(sql));

            if (statement instanceof CreateTable) {
                CreateTable createTable = (CreateTable) statement;
                // 获取表名
                String tableName = createTable.getTable().getName();
                //转义
                tableName = RedisStreamConstants.NOTA_SEARCH_PRE + tableName.replace("`", "").trim();
                IndexDefinition rule = new IndexDefinition(IndexDefinition.Type.HASH)
                        .setPrefixes(tableName + ":")
                        .setLanguage("chinese");
                Schema schema = new Schema();
                // 默认给一个 agg字段用于排序
                schema.addNumericField("agg");
                // 获取列定义
                List<ColumnDefinition> columnDefinitions = createTable.getColumnDefinitions();
                for (ColumnDefinition columnDefinition : columnDefinitions) {
                    // 获取字段名
                    String columnName = columnDefinition.getColumnName().replace("`", "").trim();
                    ColDataType colDataType = columnDefinition.getColDataType();
                    // 只有时间类型的才会是范围查询
                    if (colDataType.toString().toUpperCase().contains("DATETIME")) {
                        schema.addSortableNumericField(columnName);
                    } else {
                        schema.addTagField(columnName);
                    }
                    // System.out.println("Column Name: " + columnName + "colDataType:" + colDataType.getDataType());
                }
                UnifiedJedis client = SpringUtil.getBean(UnifiedJedis.class);
                client.ftCreate(tableName,
                        IndexOptions.defaultOptions().setDefinition(rule),
                        schema);
                isSuc = true;
            } else if (statement instanceof Alter) {
                Alter alterStatement = (Alter) statement;
                // 获取表名
                String tableName = RedisStreamConstants.NOTA_SEARCH_PRE + alterStatement.getTable().getName().replace("`", "").trim();
                //System.out.println("Table Name: " + tableName);
                // 获取修改表的表达式列表
                List<AlterExpression> alterExpressions = alterStatement.getAlterExpressions();
                Schema schema = new Schema();
                int count = 0;
                for (AlterExpression alterExpression : alterExpressions) {
                    // 获取修改的列名
                    AlterOperation operation = alterExpression.getOperation();
                    if (AlterOperation.ADD.equals(operation) || AlterOperation.CHANGE.equals(operation)) {
                        List<AlterExpression.ColumnDataType> colDataTypeList = alterExpression.getColDataTypeList();
                        for (AlterExpression.ColumnDataType columnDataType : colDataTypeList) {
                            String columnName = columnDataType.getColumnName().replace("`", "").trim();
                            ColDataType colDataType = columnDataType.getColDataType();
                            if (colDataType.toString().toUpperCase().contains("DATETIME")) {
                                schema.addSortableNumericField(columnName);
                                count++;
                            } else {
                                schema.addTagField(columnName);
                                count++;
                            }
                            //System.out.println(columnName + ":" + colDataType);
                        }
                    }
                }
                if (count > 0) {
                    SpringUtil.getBean(UnifiedJedis.class).ftAlter(tableName, schema);
                    isSuc = true;
                }
            } else {
                log.warn("当前sql不需要解析,sql:{}", sql);
            }
        } catch (Exception e) {
            dataChangeInfo.setErrInfo(e.getMessage());
            log.error("sql解析失败,sql:{}", sql, e);
        }
        return isSuc;
    }


    /**
     * 不带 try catch
     *
     * @param dataChangeInfo
     * @return
     * @throws JSQLParserException
     */
    public static void parseSql2(DataChangeInfo dataChangeInfo) throws JSQLParserException {

        String sql = dataChangeInfo.getDdl();

        Statement statement = CCJSqlParserUtil.parse(new StringReader(sql));

        if (statement instanceof CreateTable) {
            CreateTable createTable = (CreateTable) statement;
            // 获取表名
            String tableName = createTable.getTable().getName();
            //转义
            tableName = RedisStreamConstants.NOTA_SEARCH_PRE + tableName.replace("`", "").trim();
            IndexDefinition rule = new IndexDefinition(IndexDefinition.Type.HASH)
                    .setPrefixes(tableName + ":")
                    .setLanguage("chinese");
            Schema schema = new Schema();
            // 默认给一个 agg字段用于排序
            schema.addNumericField("agg");
            // 获取列定义
            List<ColumnDefinition> columnDefinitions = createTable.getColumnDefinitions();
            for (ColumnDefinition columnDefinition : columnDefinitions) {
                // 获取字段名
                String columnName = columnDefinition.getColumnName().replace("`", "").trim();
                ColDataType colDataType = columnDefinition.getColDataType();
                // 只有时间类型的才会是范围查询
                if (colDataType.toString().toUpperCase().contains("DATETIME")) {
                    schema.addSortableNumericField(columnName);
                } else {
                    schema.addTagField(columnName);
                }
                // System.out.println("Column Name: " + columnName + "colDataType:" + colDataType.getDataType());
            }
            UnifiedJedis client = SpringUtil.getBean(UnifiedJedis.class);
            client.ftCreate(tableName,
                    IndexOptions.defaultOptions().setDefinition(rule),
                    schema);
        } else if (statement instanceof Alter) {
            Alter alterStatement = (Alter) statement;
            // 获取表名
            String tableName = RedisStreamConstants.NOTA_SEARCH_PRE + alterStatement.getTable().getName().replace("`", "").trim();
            //System.out.println("Table Name: " + tableName);
            // 获取修改表的表达式列表
            List<AlterExpression> alterExpressions = alterStatement.getAlterExpressions();
            Schema schema = new Schema();
            int count = 0;
            for (AlterExpression alterExpression : alterExpressions) {
                // 获取修改的列名
                AlterOperation operation = alterExpression.getOperation();
                if (AlterOperation.ADD.equals(operation) || AlterOperation.CHANGE.equals(operation)) {
                    List<AlterExpression.ColumnDataType> colDataTypeList = alterExpression.getColDataTypeList();
                    for (AlterExpression.ColumnDataType columnDataType : colDataTypeList) {
                        String columnName = columnDataType.getColumnName().replace("`", "").trim();
                        ColDataType colDataType = columnDataType.getColDataType();
                        if (colDataType.toString().toUpperCase().contains("DATETIME")) {
                            schema.addSortableNumericField(columnName);
                            count++;
                        } else {
                            schema.addTagField(columnName);
                            count++;
                        }
                        //System.out.println(columnName + ":" + colDataType);
                    }
                }
            }
            if (count > 0) {
                SpringUtil.getBean(UnifiedJedis.class).ftAlter(tableName, schema);
            }
        } else {
            log.warn("当前sql不需要解析,sql:{}", sql);
        }

    }

}
